#!/usr/bin/python3
# -*- coding:utf-8 -*-
""""
@Author: shijianbo
@Email:
@Time: 2023/4/6 13:53
@File: moveData.py
"""
from time import time
from queue import Queue, Empty
from threading import Lock, Thread
import cx_Oracle
from Common.databases import MysqlUtil, OceanBaseUtil, OracleUtil,DaMengUtil,PostgreSqlUtil,MssqlUtil
from Common.databases.basic import DatabaseUtil
from Common.Logs import logger


class Move(object):
    def __init__(self,source:DatabaseUtil,target:DatabaseUtil,owner,table):
        # 多线程查询sql
        self.querySqlQueue=Queue()
        # 多线程插入sql
        self.insertSqlQueue=Queue()
        # 源数据库连接
        self.source=source
        # 目标数据库连接
        self.target=target
        # 用户名
        self.owner=owner
        # 表
        self.table=table
        # 临时表
        self.__tempTable= f"\"__{self.table}_TEMP\""
        # self.__tempTable= self.table
        # sql分割数
        self.SQLSPLITCOUNT = 1000
        # 阻塞
        self.BLOCK = True
        # sql参数
        self.SQLARGS = {
            'autoCommit': False
        }
        # 查询线程锁
        self.QUERYLOCK=Lock()
        # 插入线程锁
        self.INSERTLOCK=Lock()
        # 查询线程数
        self.queryThreadCount=20
        # 插入线程数
        self.insertThreadCount=10
        # 查询线程池
        self.queryThreadPool=[]
        # 插入线程池
        self.insertThreadPool=[]
        # 查询sql队列总长度
        self.queryCountTotal=0
        # 插入sql计数
        self.insertCount=0
        # 计数器LOCK
        self.countLock=Lock()
        # 生成器
        self.queryBar=self.__processQueryBar()

    def __processQueryBar(self):
        '''
        打印进度条
        '''
        start=time()
        while True:
            percent=self.insertCount/self.queryCountTotal
            done='▓'*int(percent*100//1//2)
            todo='-'*(50-int(percent*100//1//2))
            print(f"\r\033[0;31m{self.owner}.{self.table}\033[0m|\033[0;33mtotal:{self.count}\033[0m|\033[0;34m{percent*100: 6.2f}%\033[0m[\033[0;35m{done}{todo}\033[0m]\033[5;32m{time()-start:.2f}s\033[0m",end='')
            if percent>=1:
                print()
                break
            yield

    def __createTempTable(self):
        '''
        创建对应的临时表,用于存储数据
        '''
        if isinstance(self.source,OracleUtil):
            sql=f"CREATE GLOBAL TEMPORARY table {self.owner}.{self.__tempTable} ON COMMIT DELETE ROWS AS select t.*,rownum rn from {self.owner}.{self.table} t"
        else:
            logger.error('其他的源数据库待支持')
            raise Exception('其他的源数据库待支持')
        self.source.execsql(sql,**self.SQLARGS)
        # 创建主键
        sql=f"ALTER TABLE {self.owner}.{self.__tempTable} add PRIMARY KEY(RN)"
        self.source.execsql(sql, **self.SQLARGS)
        sql=f"INSERT INTO  {self.owner}.{self.__tempTable} select t.*,rownum rn from {self.owner}.{self.table} t"
        self.source.execsql(sql,**self.SQLARGS)
        logger.debug(f"临时表{self.owner}.{self.__tempTable}已新建并插入数据")

    def __dropTempTable(self):
        '''
        删除对应临时表
        '''
        sql=f"DROP TABLE {self.owner}.{self.__tempTable}"
        self.source.execsql(sql,**self.SQLARGS)
        logger.debug(f"临时表{self.owner}.{self.__tempTable}已删除")

    def __truncateTable(self):
        '''
        truncate目标表
        '''
        sql=f"truncate table {self.owner}.{self.table};"
        self.target.execsql(sql,**self.SQLARGS)
        logger.debug(f"目标表{self.owner}.{self.table}已truncate")

    def __splitSql(self):
        '''
        将查询的表根据数量分页查询,插入队列
        '''
        sql=f"select count(*) as COUNT from {self.owner}.{self.__tempTable}"
        self.count=self.source.execsql(sql,**self.SQLARGS)[0]['COUNT']
        for front,behind in [(self.SQLSPLITCOUNT * (i - 1) + 1, self.SQLSPLITCOUNT * i) for i in range(1, self.count // self.SQLSPLITCOUNT + 2)]:
            sql=f"select * from {self.owner}.{self.__tempTable} where rn<={behind} and rn>={front}"
            self.querySqlQueue.put(sql)
        else:
            logger.debug(f"查询sql队列已就绪")
            # 查询sql队列长度
            self.queryCountTotal=self.querySqlQueue.qsize()
            next(self.queryBar)

    def __getQuerySqlQueueLock(self):
        '''
        获取查询sql,获取队列元素前加锁,保证线程安全,取得元素后释放锁
        '''
        self.QUERYLOCK.acquire(blocking=True,timeout=-1)
        try:
            if not self.querySqlQueue.empty():
                return self.querySqlQueue.get(block=False)
            else:
                return False
        finally:
            self.QUERYLOCK.release()

    def __getInsertSqlQueueLock(self):
        '''
        获取插入sql,获取队列元素前加锁,保证线程安全,取得元素后释放锁
        '''
        self.INSERTLOCK.acquire(blocking=True,timeout=-1)
        try:
            return self.insertSqlQueue.get(block=self.BLOCK)
        except Empty as e:
            return False
        finally:
            self.INSERTLOCK.release()

    def __query(self):
        '''
        从源查询数据,生成sql,放入队列
        '''
        while True:
            sql=self.__getQuerySqlQueueLock()
            if not sql:
                break
            data=self.source.execsql(sql,**self.SQLARGS)
            cols=','.join([key for key in data[0] if key not in ('ROWNUM','RN','rownum','rn')])
            insertSql=f"insert into {self.owner}.{self.table} ({cols}) values"
            for _data in data:
                values=''
                for k,v in _data.items():
                    if k in ('ROWNUM','RN','rownum','rn'):
                        continue
                    if isinstance(v,str):
                        _v = v.replace("'", "''")
                        values+=f"'{_v}'"
                    elif isinstance(v, (int, float)):
                        values += f'{v}'
                    elif isinstance(v, cx_Oracle.LOB):
                        _v = v.read().replace("'", "''")
                        values += f"'{_v}'"
                    elif v == None:
                        values += 'null'
                    else:
                        values += f"'{v}'"
                    values+=','
                else:
                    # 去除最后一个,
                    values=values[:-1]
                    insertSql+=f"({values}),"
            else:
                # 去除最后一个,
                insertSql=insertSql[:-1]+';'
                self.insertSqlQueue.put(insertSql)

    def __insert(self):
        '''
        从队列获取sql,插入目标数据库
        '''
        while True:
            sql=self.__getInsertSqlQueueLock()
            if not sql:
                break
            try:
                self.target.execsql(sql,**self.SQLARGS)
                self.countLock.acquire(blocking=True,timeout=-1)
                self.insertCount+=1
                try:
                    next(self.queryBar)
                except StopIteration:
                    pass
                self.countLock.release()
            except Exception as e:
                raise e

    def __createThread(self):
        '''
        创建查询和插入的线程
        '''
        for i in range(self.queryThreadCount):
            self.queryThreadPool.append(Thread(target=self.__query))
        for i in range(self.insertThreadCount):
            self.insertThreadPool.append(Thread(target=self.__insert))

    def __startThread(self):
        for i in self.queryThreadPool:
            i.start()
        for i in self.insertThreadPool:
            i.start()

    def __joinThread(self):
        for i in self.queryThreadPool:
            i.join()
        else:
            logger.debug('源表查询已完毕,等待插入完成')
            self.__dropTempTable()
            self.BLOCK=False
            self.insertSqlQueue.put(False)
            self.source.commit()
        for i in self.insertThreadPool:
            i.join()
        else:
            logger.debug('目标表插入已完成')
            self.target.commit()

    def run(self):
        start=time()
        self.__createTempTable()
        self.__truncateTable()
        self.__splitSql()
        self.__createThread()
        self.__startThread()
        self.__joinThread()
        logger.info(f'{self.table}已经迁移完成,数据量:{self.count},总耗时:{time()-start:.2f}s')


if __name__ == '__main__':
    with OracleUtil('sys','sys','10.20.46.181',server='orclcdb') as source:
        with DaMengUtil('SYSDBA','SYSDBA','10.20.249.125') as target:
            Move(source,target,'hsinfo','INFO_SHAREHOLDER_COUNT').run()
            # Move(source,target,'hsrcce','RCCE_RISK').run()

